package com.poetic.network.flow.watermark;

import com.poetic.network.flow.domain.SensorReading;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;

public class MyAssigner implements AssignerWithPeriodicWatermarks<SensorReading> {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
    // 允许延迟1秒
    private int bound = 1000;
    // 定义变量接收最大值，因为水位线只能递增，不能变小
    private Long maxTs = Long.MIN_VALUE;
    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(maxTs - bound);
    }

    @Override
    public long extractTimestamp(SensorReading element, long recordTimestamp) {
        Long timestamp = element.getTimestamp();
        maxTs =  Math.max(timestamp, maxTs);

        System.out.println("传入数据:" + timestamp + "|" + sdf.format(timestamp) + ",此时的水印为:" + getCurrentWatermark().getTimestamp() + "|" + sdf.format(getCurrentWatermark().getTimestamp()));

        // 秒转为毫秒
        return timestamp;
    }
}
